package com.study.kafka.custom.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * 生产者拦截器
 * 顺序：拦截器->序列化器->分区器
 * 功能：过滤数据，修改数据内容等
 * 使用：props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, DemoProducerInterceptor.class.getName())
 *
 * @author: stephen.shen
 * @create: 2019-06-18 11:25
 */
public class DemoProducerInterceptor implements ProducerInterceptor<String, String> {
    private int successCounter = 0;
    private int errorCounter = 0;

    /**
     * 消息序列化和计算分区前调用
     *
     * @param record
     * @return
     */
    @Override
    public ProducerRecord onSend(ProducerRecord<String, String> record) {
        // 在value之间加时间戳
        return new ProducerRecord(record.topic(),
                record.partition(),
                record.key(),
                System.currentTimeMillis() + ":" + record.value());
    }

    /**
     * 消息被应答前或消息发送失败时调用,
     * 优先于用户设定的 Callback 之前执行.
     *
     * @param metadata
     * @param exception
     */
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // 记录发送成功和失败次数
        if (exception == null) {
            successCounter++;
        } else {
            errorCounter++;
        }
    }

    @Override
    public void close() {
        // 保存结果
        System.out.println("Successful sent: " + successCounter);
        System.out.println("Failed sent: " + errorCounter);
    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
